package com.honeybee.cloud.framework.rocket.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Objects;

/**
 * @Description 基础生产者
 * @Author DELL
 * @Create 2019-04-04
 * @Since 1.0.0
 */
@Slf4j
public abstract class AbstractProducer implements InitializingBean {

    /**
     * 生产者
     */
    protected DefaultMQProducer producer;

    /**
     * 生产者组
     */
    protected String producerGroup = "DEFAULT_PRODUCER_GROUP";

    /**
     * mq连接路径
     */
    protected String nameSrvAddress;

    /**
     * 初始化
     *
     * @throws Exception
     */
    @Override
    public abstract void afterPropertiesSet() throws Exception;

    /**
     * 开启生产组
     *
     * @param defaultMQProducer
     * @throws MQClientException
     */
    protected void startProducerGroup(DefaultMQProducer defaultMQProducer) throws MQClientException {
        if (Objects.nonNull(defaultMQProducer)) {
            this.producer = defaultMQProducer;
            return;
        }
        log.info("开始启动生产组:{},地址:{}", producerGroup, nameSrvAddress);
        Assert.hasLength(nameSrvAddress, "nameSrvAddress is not null");
        defaultMQProducer = new DefaultMQProducer(producerGroup);
        defaultMQProducer.setNamesrvAddr(nameSrvAddress);
        defaultMQProducer.setVipChannelEnabled(false);
        StringBuffer sdf = new StringBuffer();
        sdf.append(producerGroup).append("_").append(System.currentTimeMillis()).append("_").append(RandomUtils.nextLong());
        String instanceName = sdf.toString();
        defaultMQProducer.setInstanceName(instanceName);
        defaultMQProducer.start();
        log.info("成功启动生产组:{},实例名:{}", producerGroup, instanceName);
        this.producer = defaultMQProducer;
    }

    /**
     * 使用时请调用此方法实例化，完成相关配置
     *
     * @param nameSrvAddress
     * @param producerGroup
     */
    protected void init(String nameSrvAddress, String producerGroup) {
        this.nameSrvAddress = nameSrvAddress;
        this.producerGroup = producerGroup;
    }


    /**
     * 使用时请调用此方法实例化，完成相关配置
     *
     * @param producer
     */
    protected void init(DefaultMQProducer producer) {

        this.producer = producer;
    }

    /**
     * 开启生产组
     *
     * @throws MQClientException
     */
    protected void startProducerGroup() throws MQClientException {
        startProducerGroup(this.producer);
        Assert.notNull(producer, "producer is null");
    }

    /**
     * 发送顺序消息
     *
     * @param topic
     * @param tags
     * @param message
     * @param key
     * @param delayTimeLevel
     * @param hashKey
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult syncSendOrderly(String topic, String tags, String message, String key, Integer delayTimeLevel, String hashKey, long timeout)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        Assert.hasLength(topic, "topic is not null");
        Assert.hasLength(message, "message is not null");
        Assert.hasLength(tags, "tags is not null");
        log.info("发送消息:{},{},{}", topic, tags, message);
        Message msg = new Message(topic,
                StringUtils.defaultString(tags),
                StringUtils.defaultString(key),
                (message).getBytes(RemotingHelper.DEFAULT_CHARSET));
        if (Objects.nonNull(delayTimeLevel)) {
            msg.setDelayTimeLevel(delayTimeLevel);
        }
        SendResult sendResult = producer.send(msg, (List<MessageQueue> mgs, Message messages, Object arg) -> {
            int id = Objects.hashCode(hashKey);
            if (id < 0) {
                id = 0;
            }
            int index = id % mgs.size();
            return mgs.get(index);
        }, StringUtils.defaultString(key), timeout);
        log.info("发送消息:{},{},{},结果:{}", topic, tags, message, sendResult);
        return sendResult;
    }

    /**
     * 发送顺序消息
     *
     * @param topic
     * @param tags
     * @param message
     * @param key
     * @param delayTimeLevel
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult syncSendOrderly(String topic, String tags, String message, String key, Integer delayTimeLevel)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        return syncSendOrderly(topic, tags, message, key, delayTimeLevel, null, 3000);
    }

    /**
     * 发送顺序消息
     *
     * @param topic
     * @param tags
     * @param message
     * @param key
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    protected SendResult syncSendOrderly(String topic, String tags, String message, String key)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        return syncSendOrderly(topic, tags, message, key, null, null, 3000);
    }

    /**
     * 发送顺序消息
     *
     * @param topic
     * @param tags
     * @param message
     * @param key
     * @param timeout
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult syncSendOrderly(String topic, String tags, String message, String key, long timeout)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        return syncSendOrderly(topic, tags, message, key, null, null, timeout);
    }

    /**
     * 发送顺序消息
     *
     * @param topic
     * @param tags
     * @param message
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult syncSendOrderly(String topic, String tags, String message)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        return syncSendOrderly(topic, tags, message, null, null, null, 3000);
    }

    /**
     * 发送顺序消息
     *
     * @param topic
     * @param tags
     * @param message
     * @param timeout
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult syncSendOrderly(String topic, String tags, String message, long timeout)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        return syncSendOrderly(topic, tags, message, null, null, null, timeout);
    }

    /**
     * 发送顺序消息
     *
     * @param topic
     * @param message
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult syncSendOrderly(String topic, String message)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        return syncSendOrderly(topic, null, message, null, null, null, 3000);
    }

    /**
     * 发送顺序消息
     *
     * @param topic
     * @param message
     * @param timeout
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult syncSendOrderly(String topic, String message, long timeout)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        return syncSendOrderly(topic, null, message, null, null, null, timeout);
    }

    /**
     * 同步发送非顺序消息
     *
     * @param topic
     * @param tags
     * @param message
     * @param key
     * @param timeout
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult syncSend(String topic, String tags, String message, String key, long timeout)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        Assert.hasLength(topic, "topic is not null");
        Assert.hasLength(message, "message is not null");
        log.info("发送消息:{},{},{}", topic, tags, message);
        Message msg = new Message(topic,
                StringUtils.defaultString(tags),
                StringUtils.defaultString(key),
                (message).getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult sendResult = producer.send(msg, timeout);
        log.info("发送消息:{},{},{},结果:{}", topic, tags, message, sendResult);
        return sendResult;
    }

    /**
     * 同步发送非顺序消息
     *
     * @param topic
     * @param tags
     * @param message
     * @param key
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult syncSend(String topic, String tags, String message, String key)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        return syncSend(topic, tags, message, key, 0);
    }

    /**
     * 同步发送非顺序消息
     *
     * @param topic
     * @param tags
     * @param message
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult syncSend(String topic, String tags, String message)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        return syncSend(topic, tags, message, null, 3000);
    }

    /**
     * 同步发送非顺序消息
     *
     * @param topic
     * @param message
     * @return
     * @throws UnsupportedEncodingException
     * @throws MQClientException
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult syncSend(String topic, String message)
            throws UnsupportedEncodingException, MQClientException, MQBrokerException, RemotingException, InterruptedException {
        return syncSend(topic, null, message, null, 3000);
    }
}
